package com.iu.distributed.lock.zk;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.locks.Lock;

import org.I0Itec.zkclient.ZkClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Preconditions;
import com.google.common.base.Strings;

public abstract class ZkDistributedLockParent implements Lock {
	private final static Logger LOG = LoggerFactory.getLogger(ZkDistributedLockParent.class);
	/**
	 * zk客户端集合
	 */
	private static Map<String, ZkClient> zkClientMap = new HashMap<String, ZkClient>();
	/**
	 * zk客户端关联实例
	 */
	private static Map<ZkClient,Set<ZkDistributedLockParent>> zkReference = new HashMap<ZkClient,Set<ZkDistributedLockParent>>();
	/**
	 * 初始化锁
	 */
	private Object initLock = new Object();
	/**
	 * zk客户端
	 */
	private ZkClient client;
	/**
	 * zk连接配置
	 */
	private IZkLockConfig config;

	public ZkDistributedLockParent(IZkLockConfig config) {
		validteConfig(config);
		if (!zkClientMap.containsKey(config.getZkServers())) {
			synchronized (initLock) {
				if (!zkClientMap.containsKey(config.getZkServers())) {
					LOG.debug("初始化zookeeper连接,zkservers:{},zkconnectTimeout:{}",config.getZkServers(),config.getZkConnectTimeout());
					client = new ZkClient(config.getZkServers(),
							config.getZkConnectTimeout() != null ? config.getZkConnectTimeout() : Integer.MAX_VALUE);
					zkClientMap.put(config.getZkServers(), client);
				}else{
					LOG.debug("zookeeper连接已存在从池中获取,zkservers:{},zkconnectTimeout:{}",config.getZkServers(),config.getZkConnectTimeout());
				}
			}
		}else{
			LOG.debug("zookeeper连接已存在从池中获取,zkservers:{},zkconnectTimeout:{}",config.getZkServers(),config.getZkConnectTimeout());
		}
		if(!zkReference.containsKey(client)){
			synchronized (initLock) {
				if(!zkReference.containsKey(client)){
					zkReference.put(client, new HashSet<ZkDistributedLockParent>());
				}
			}
		}
		zkReference.get(client).add(this);
		client = zkClientMap.get(config.getZkServers());
		this.config = config;
	}

	private void validteConfig(IZkLockConfig config) {
		Preconditions.checkNotNull(config, "ZkDistributedLock 配置信息不能为空");
		Preconditions.checkArgument(!Strings.isNullOrEmpty(config.getZkServers()),
				"ZkDistributedLock 需指定zookeeper服务地址");
	}
	/**
	 * 
	 * 获取zookeeper 客户端
	 * @return ZkClient      
	 */
	protected ZkClient getZkClient() {
		return client;
	}
	/**
	 * 
	 * 获取zk连接配置
	 * @return IZkLockConfig      
	 */
	protected IZkLockConfig getConfig() {
		return config;
	}
	/**
	 * 
	 * 销毁对象，移除当前实例与zk客户端的关联，如zk客户端无其它关联则关闭并移除zk客户端;
	 * @return void      
	 */
	public void destory(){
		LOG.debug("zookeeper连接销毁,zkservers:{},zkconnectTimeout:{}",config.getZkServers(),config.getZkConnectTimeout());
		zkReference.get(client).remove(this);
		if(zkReference.get(client).isEmpty()){
			synchronized (initLock) {
				if(zkReference.get(client).isEmpty()){
					zkClientMap.remove(client);
					zkReference.remove(client);
					client.close();
				}
			}
		}
	}
}
